package com.sia.springcloud.process.consumer;

import com.alibaba.fastjson.JSONObject;
import com.sia.springcloud.process.influxDB.InfluxDbTool;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.TimeUnit;


/**
 * @ClassName KafkaConsumerAbb
 * @Description TODO
 * @Author sgw
 * @Date 2022-4-16 12:55
 * @Version 1.0
 **/
public class KafkaConsumerS7 implements Runnable {


    private KafkaConsumer<String, String> consumer;
    private ConsumerRecords<String, String> msgList;
    private InfluxDbTool influxDbTool;
    private  String topic;
    private static final String GROUPID = "etl";


    public KafkaConsumerS7(String topicName, InfluxDbTool influxDbTool) {
        this.topic = topicName;
        this.influxDbTool=influxDbTool;
        init();
    }

    @Override
    public void run() {
        System.out.println("---------开始消费---------");
        List<String> list=new ArrayList<String>();
        List<Long> list2=new ArrayList<Long>();
        BatchPoints batchPoints  = BatchPoints.builder().build();
        Map<String,String> tempMap=new HashMap<String,String>();
        while (true) {
            try {
                msgList = consumer.poll(100);
                if(null!=msgList){
                    //s7是以plc为单位存入kafka的
                  for (ConsumerRecord<String, String> record : msgList) {
                        list.add(record.value());
                        list2.add(record.offset());
                        System.out.println("主题:"+ record.topic());
                        if(record.topic().equals("s7") && null != record && record.value().contains("{")){
                            JSONObject jsonObject= JSONObject.parseObject(record.value());



                            Point point = Point.measurement(jsonObject.getString("name"))
                                    .addField("collectTime",jsonObject.getString("time"))
                                    .tag("variableValue",jsonObject.getString("value"))
                                    .tag("deviceName",jsonObject.getString("name"))
                                    .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS).build();
                            batchPoints.point(point);//将point放入batch中

                        }
                    }
//                  try{
                      influxDbTool.saveInfluxDb(batchPoints,"s7");
                      consumer.commitSync();
                      batchPoints.getPoints().clear();
                      list.clear();
                      list2.clear();
                }else{
                    Thread.sleep(1000);
                }
            }catch (Exception e){
                //logger.error("s7存入influxdb失败:"+e.getMessage());
            }
        }
    }




    private void init() {
        Properties props = new Properties();
        //kafka消费的的地址
        props.put("bootstrap.servers", "127.0.0.1:9092");
        //组名 不同组名可以重复消费
        props.put("group.id", GROUPID);
        //是否自动提交
        props.put("enable.auto.commit", "false");
        //超时时间
        props.put("session.timeout.ms", "30000");
        //一次最大拉取的条数
        props.put("max.poll.records", 5000);
//		earliest当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
//		latest
//		当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
//		none
//		topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
        props.put("auto.offset.reset", "earliest");
        //序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        //订阅主题列表topic
        this.consumer.subscribe(Arrays.asList(topic));

        System.out.println("s7初始化!");
    }
}
